#ifndef LSV_WBUFFER_INTERNAL_H
#define LSV_WBUFFER_INTERNAL_H

#include "vec.h"
#include "lsv_wbuffer.h"

#define WAL_HEAD_MAGIC   0x33333333
#define WAL_HEAD_VERSION 1

#pragma pack(8)

// wbuf_page_id: 7
typedef struct {
        uint32_t magic;
        uint32_t version;
        uint64_t last_lsn;    // 提交LSN
        uint32_t head_chunk;  // 未提交的segment第一个chunk id
} checkpoint_t;

// page_id: 8+，记录wal所用chunk信息
typedef struct {
        uint32_t head_len;
        uint32_t crc;
        uint32_t magic;
        uint32_t version;

        uint32_t segment_num;
        uint32_t segment_len;
        uint32_t chunk_num;
        uint32_t chunk_ids[0];
} segment_head_t;

typedef struct {
        uint64_t off;
        uint32_t size;
        uint32_t chunk_idx;
        uint32_t chunk_id;
        uint32_t page_idx;
        uint32_t page_num;
} wal_io_frag_t;

typedef struct {
        // meta
        uint32_t head_len;
        uint32_t crc;
        uint32_t magic;
        uint32_t version;

        // io
        uint64_t lsn;
        uint64_t lba;
        uint32_t size;
        uint32_t frag_num;

        // TODO 跨chunk会划分成IO片段
        wal_io_frag_t frags[IO_FRAG_MAX];
} wal_io_head_t;

#pragma pack()


static inline int _chunk_is_full(lsv_wbuf_wal_chunk_t *wchunk) {
        return wchunk->next_page == LSV_WBUF_PAGE_NUM;
}

/**
 * @pre offset是页对齐的
 *
 * @param size
 * @return
 * @see lsv_page_num
 */
static inline uint32_t _page_num(uint32_t size) {
        return size / LSV_PAGE_SIZE + (size % LSV_PAGE_SIZE ? 1 : 0);
}

static inline uint32_t wal_io_head_len() {
        return sizeof(wal_io_head_t);
}

/**
 * @file
 *
 * - wbuffer/internal
 * - IO: buffer(+index)
 * - WAL
 * - load
 */

int lsv_wbuffer_mem_init(lsv_volume_proto_t *lsv_info);

int lsv_wbuffer2log(lsv_volume_proto_t *lsv_info, lsv_chunk_buf_t *chunk_buf);

/**
 * IO
 *
 * @param chunk_buf tail chunk or NULL
 */
int lsv_wbuffer_do_flush(lsv_volume_proto_t * lsv_info);

// -- buffer manager

#if 0

int lsv_wbuffer_segment_init(lsv_volume_proto_t *lsv_info);
int lsv_wbuffer_segment_destroy(lsv_volume_proto_t *lsv_info);

/**
 * @brief malloc, and add to work list
 *
 * @param lsv_info
 * @return
 */
int lsv_wbuffer_segment_malloc(lsv_volume_proto_t *lsv_info);

/**
 * @brief get current write point
 *
 * @param lsv_info
 * @return
 */
lsv_wbuf_segment_t *get_current_segment(lsv_volume_proto_t * lsv_info);

int lsv_wbuffer_segment_get_free_size(lsv_wbuf_segment_t *segment);

int lsv_wbuffer_segment_check(lsv_volume_proto_t *lsv_info,
                              lsv_io_opt_t *io_opt,
                              lsv_wbuf_segment_t **segment);

int lsv_wbuffer_segment_check_with_malloc(lsv_volume_proto_t *lsv_info,
                                          lsv_io_opt_t *io_opt,
                                          lsv_wbuf_segment_t **segment);

int lsv_wbuffer_segment_append_page(lsv_volume_proto_t *lsv_info, lsv_io_opt_t *io_opt, buffer_t *page_buf);

int lsv_wbuffer_segment_gc(lsv_volume_proto_t *lsv_info);
int lsv_wbuffer_segment_commit(lsv_volume_proto_t *lsv_info, lsv_chunk_buf_t *chunk, int in_use);

int lsv_wbuffer_segment_push_log(lsv_volume_proto_t *lsv_info, lsv_chunk_buf_t *chunk_buf, int flush);

// index
int lsv_wb_hash_index_init(lsv_volume_proto_t *lsv_info);
int lsv_wb_hash_index_destroy(lsv_volume_proto_t *lsv_info);

int lsv_wb_hash_index_insert(lsv_volume_proto_t *volume_protom, lsv_wbuf_segment_t * wbuf_chunk_block,
                                   lsv_u64_t lba, lsv_u32_t chunk_id, lsv_u32_t chunk_page_idx);
int lsv_wb_hash_index_gc(lsv_wbuf_segment_t * chunk_block);

int lsv_wb_hash_index_lookup(index_block_head *hash, lsv_u64_t offset, lsv_u32_t size, buffer_t *append_buf);

#endif

// wbuf chunk-level
int lsv_wbuffer_chunk_init(lsv_volume_proto_t *lsv_info);
int lsv_wbuffer_chunk_destroy(lsv_volume_proto_t *lsv_info);

int lsv_wbuffer_chunk_append_page(lsv_volume_proto_t *lsv_info, lsv_io_opt_t *io_opt, buffer_t *page_buf);
int lsv_wbuffer_chunk_push_log(lsv_volume_proto_t *lsv_info, lsv_wbuf_chunk_t *chunk, int flush, int is_prepare);

int lsv_wbuffer_chunk_commit(lsv_volume_proto_t *lsv_info, lsv_chunk_buf_t *chunk_buf, int in_use);
int lsv_wbuffer_chunk_gc(lsv_volume_proto_t *lsv_info);

lsv_wbuf_chunk_t *get_current_chunk(lsv_volume_proto_t *lsv_info);
int lsv_wbuffer_chunk_get_free_page_number(lsv_volume_proto_t *lsv_info);

int lsv_wbuffer_chunk_check(lsv_volume_proto_t *lsv_info, uint64_t off, uint32_t size);
int lsv_wbuffer_chunk_expand(lsv_volume_proto_t *lsv_info);
int lsv_wbuffer_chunk_ensure(lsv_volume_proto_t *lsv_info, uint64_t off, uint32_t size);

// IO-level ops
int lsv_wbuffer_chunk_prepare(lsv_volume_proto_t *lsv_info, const io_t *io, buffer_t *buf, wbuf_io_head_t **head);
int lsv_wbuffer_chunk_fill(lsv_volume_proto_t *lsv_info, wbuf_io_head_t *head);
int lsv_wbuffer_chunk_fill_v2(lsv_volume_proto_t *lsv_info, wbuf_io_head_t *head);
int lsv_wbuffer_chunk_append_page_just_do_it(wbuf_io_frag_t *io_frag, buffer_t *buf);

int lsv_page_write(wbuf_io_frag_t *io_frag);

// index
int lsv_wb_hash_index_init_v2(lsv_volume_proto_t *lsv_info);
int lsv_wb_hash_index_destroy_v2(lsv_volume_proto_t *lsv_info);

int lsv_wb_hash_index_insert_v2(lsv_volume_proto_t *volume_protom,
                                uint64_t lba,
                                lsv_wbuf_chunk_t *chunk,
                                lsv_u32_t chunk_page_idx);
int lsv_wb_hash_index_gc_v2(lsv_wbuf_chunk_t *chunk);
int lsv_wb_hash_index_clean_all(lsv_volume_proto_t *lsv_info);

int lsv_wb_hash_index_lookup_v2(index_block_head *hash, lsv_u64_t offset, lsv_u32_t size, buffer_t *append_buf);

// WAL manager

int _load_chunk_page(lsv_volume_proto_t *lsv_info, uint32_t chunk_id, uint32_t off, uint32_t size, lsv_s8_t *buf);
int _load_chunk(lsv_volume_proto_t *lsv_info, uint32_t chunk_id, void **_buf);

int _encode_checkpoint(lsv_volume_proto_t *lsv_info, uint32_t head_chunk, uint64_t last_lsn);
int _decode_checkpoint(lsv_volume_proto_t *lsv_info, checkpoint_t *_cp);

int read_chunk_list(lsv_volume_proto_t *lsv_info, segment_head_t **_seg_head);

// wal2 segment mode (NOW use this mode, 支持正常写入，或螺旋写入)
int lsv_wal2_segment_init(lsv_volume_proto_t *lsv_info);
int lsv_wal2_segment_destroy(lsv_volume_proto_t *lsv_info);

// int lsv_wal2_segment_check_with_malloc(lsv_volume_proto_t *lsv_info, int size, lsv_wbuf_wal_segment_t **_segment);
int lsv_wal2_segment_check(lsv_volume_proto_t *lsv_info, int size, lsv_wbuf_wal_segment_t **segment);
int lsv_wal2_segment_ensure_free_list(lsv_volume_proto_t *lsv_info, int n);

int lsv_wal2_segment_append_prepare(lsv_volume_proto_t *lsv_info, const io_t *io, wal_io_head_t *ctx);
int lsv_wal2_segment_append(lsv_volume_proto_t *lsv_info, char *buffer);

int lsv_wal2_segment_gc(lsv_volume_proto_t *lsv_info, uint64_t lsn, uint64_t *commit_lsn);

int lsv_wal2_segment_commit(lsv_volume_proto_t *lsv_info, uint64_t lsn);

int lsv_wal2_segment_load(lsv_volume_proto_t *lsv_info);

int update_bitmap(lsv_volume_proto_t *lsv_info, lsv_u32_t log_chkid, const lsv_log_proto_t *log_proto);

int chunk_post_commit(lsv_volume_proto_t *lsv_info, lsv_chunk_buf_t *chunk, int in_use);

#if 0

// wal version 1
int lsv_wal_malloc(lsv_volume_proto_t * lsv_info, lsv_wbuf_segment_t * chunk_block);
int lsv_wal_free(lsv_volume_proto_t * lsv_info, lsv_wbuf_segment_t * chunk_block);

int lsv_wal_malloc_i(lsv_volume_proto_t *lsv_info, lsv_wbuf_segment_t * chunk_block);

int lsv_wal_append_io(lsv_volume_proto_t *lsv_info, const io_t *io, const buffer_t *buf);
int lsv_wal_append_end(lsv_volume_proto_t *lsv_info);

// wal version 2
int lsv_wal2_init(lsv_volume_proto_t *lsv_info);
int lsv_wal2_destroy(lsv_volume_proto_t *lsv_info);

int lsv_wal2_malloc(lsv_volume_proto_t *lsv_info, uint32_t count);
int lsv_wal2_free(lsv_volume_proto_t *lsv_info, int size);

int lsv_wal2_append_prepare(lsv_volume_proto_t *lsv_info, const io_t *io, wal_ioctx_t *ioctx);
int lsv_wal2_append(lsv_volume_proto_t *lsv_info, char *buffer);
int lsv_wal2_append_async(lsv_volume_proto_t *lsv_info, char *buffer);

int lsv_wal2_load(lsv_volume_proto_t *lsv_info);

int lsv_wal2_commit(lsv_volume_proto_t *lsv_info, uint64_t lsn);
int lsv_wal2_checkpoint(lsv_volume_proto_t *lsv_info);

#endif

// from align_io.c
int lba_kv_init(hashtable_t *_ht);
int lba_kv_destroy(hashtable_t ht);

int lba_kv_insert(hashtable_t ht, co_func_t func, wbuf_io_frag_t *io_frag, int need_free);

// from commit_order.c
int lba_fifo_init(hashtable_t *_ht);
int lba_fifo_destroy(hashtable_t ht);

int lba_fifo_insert_page(hashtable_t ht, wbuf_io_frag_t *io_frag);
int lba_fifo_fill_page(hashtable_t ht, co_func_t func, wbuf_io_frag_t *io_frag);

// from lock_table.c
// int ltable_init(hashtable_t *ht);
// int ltable_destroy(hashtable_t ht);

// int ltable_wrlock(hashtable_t ht, uint64_t lba);
// int ltable_rdlock(hashtable_t ht, uint64_t lba);
// int ltable_unlock(hashtable_t ht, uint64_t lba);


#endif
